package com.lee.mq.mqserver.datacenter;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: lishuo
 * Date: 2023-10-23
 * Time: 9:55
 */

import com.lee.mq.common.BinaryTool;
import com.lee.mq.common.MqException;
import com.lee.mq.mqserver.core.MSGQueue;
import com.lee.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * 通过这个类, 来针对硬盘上的消息进行管理
 */
public class MessageFileManager {
    // Stat 内部类：用于表示该队列的统计信息.
    public static class Stat {
        public int totalCount; // 总消息数量
        public int validCount; // 有效消息数量
    }

    // 定义一个初始化方法（暂不需要，便于拓展）
    public void init() {

    }

    // ----------------------- 获取消息文件相关路径 ---------------------------
    // 获取指定队列消息文件所在路径
    private String getQueueDirPath(String queueName) {
        return "./data/"+queueName;
    }
    // 获取指定对列统计文件所在路径
    private String getQueueStatPath(String queueName) {
        return getQueueDirPath(queueName)+"/queue_stat.txt";
    }
    // 获取指定队列数据文件所在路径
    private String getQueueDataPath(String queueName) {
        return getQueueDirPath(queueName)+"/queue_data.txt";
    }

    // ----------------------- 队列消息文件目录操作 ---------------------------
    // 检查队列对应的目录和文件是否存在
    public boolean checkFilesExits(String queueName) {
        // 判定队列对应的统计文件和数据文件分别是否存在
        File statFile = new File(getQueueStatPath(queueName));
        if (!statFile.exists()) {
            return false;
        }
        File dataFile = new File(getQueueDataPath(queueName));
        if (!dataFile.exists()) {
            return false;
        }
        return true;
    }

    // 创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException, MqException {
        // 1.创建队列对应的消息目录
        File msgDir = new File(getQueueDirPath(queueName));
        if (!msgDir.exists()) {
            boolean res = msgDir.mkdirs();
            if (!res) {
                throw new IOException("创建目录失败: msgDir = "+msgDir.getAbsolutePath());
            }
        }
        // 2.创建统计文件
        File statFile = new File(getQueueStatPath(queueName));
        if (!statFile.exists()) {
            boolean res = statFile.createNewFile();
            if (!res) {
                throw new IOException("创建统计文件失败: statFile = "+statFile.getAbsolutePath());
            }
        }
        // 3.创建数据文件
        File dataFile = new File(getQueueDataPath(queueName));
        if (!dataFile.exists()) {
            boolean res = dataFile.createNewFile();
            if (!res) {
                throw new IOException("创建数据文件失败: dataFile = "+dataFile.getAbsolutePath());
            }
        }
        // 4.初始化统计文件（0\t0）
        Stat stat = new Stat();
        stat.validCount = 0;
        stat.totalCount = 0;
        writeStat(queueName,stat);
    }

    // 删除队列对应的文件和目录
    public void destoryQueueFiles(String queueName) throws IOException {
        // 1.先删除文件
        File statFile = new File(getQueueStatPath(queueName));
        if (statFile.exists()) {
            boolean res1 = statFile.delete();
            if (!res1) {
                throw new IOException("删除统计文件失败: statFile = "+ statFile.getAbsolutePath());
            }
        }
        File dataFile = new File(getQueueDataPath(queueName));
        if (dataFile.exists()) {
            boolean res2 = dataFile.delete();
            if (!res2) {
                throw new IOException("删除数据文件失败: dataFile = "+ dataFile.getAbsolutePath());
            }
        }
        // 2.后删除目录
        File msgDir = new File(getQueueDirPath(queueName));
        if (msgDir.exists()) {
            boolean res3 = msgDir.delete();
            if (!res3) {
                throw new IOException("删除目录失败: msgDir = "+ msgDir.getAbsolutePath());
            }
        }
    }

    // ----------------------- 队列对应统计文件读写操作 ---------------------------
    // 读取队列对应的stat文件
    private Stat readStat(String queueName) throws MqException {
        // 1.判断文件是否存在
        File statFile = new File(getQueueStatPath(queueName));
        if (!statFile.exists()) {
            throw new MqException("文件不存在: statFile = "+ statFile.getAbsolutePath());
        }
        // 2.将stat文件信息读取到Stat对象中
        Stat stat = new Stat();
        try(InputStream inputStream = new FileInputStream(statFile)) {
            Scanner scanner  = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return stat;
    }
    // 写入队列对应的stat文件
    private void writeStat(String queueName,Stat stat) throws MqException {
        // 1.判断文件是否存在
        File statFile = new File(getQueueStatPath(queueName));
        if (!statFile.exists()) {
            throw new MqException("文件不存在: statFile = "+ statFile.getAbsolutePath());
        }
        // 2.将stat对象中的数据写入到stat文件中
        // 这里是覆盖写入
        try(OutputStream outputStream = new FileOutputStream(statFile)) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // ----------------------- 消息在文件中的读取、写入、删除操作(stat文件+data文件) ---------------------------
    // 将一条新消息写入到文件中
    public void sendMessage(MSGQueue msgQueue, Message message) throws MqException, IOException {
        // 1.判断队列对应的文件目录是否存在
        if (!checkFilesExits(msgQueue.getName())) {
            throw new MqException("队列对应的文件或目录不存在: queueName = "+msgQueue.getName());
        }
        // 2.将message对象序列化，转成二进制字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);
        // 修改写入操作：
        synchronized (msgQueue) {
            // 3.获取到queue_data.txt文件
            File dataFile = new File(getQueueDataPath(msgQueue.getName()));
            // 4.获取当前文件的长度，计算offsetBeg/offsetEnd
            long length = dataFile.length();
            message.setOffsetBeg(length+4);
            message.setOffsetEnd(length+4+messageBinary.length);
            // 5.写入消息到文件（此处是追加写，并且根据约定先写4个字节表示消息长度）
            try(OutputStream outputStream = new FileOutputStream(dataFile,true)) {
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // DataOutPutStream 可以保证writeInt()写入的是4个字节
                    dataOutputStream.writeInt(messageBinary.length);
                    // 写入消息主体
                    dataOutputStream.write(messageBinary);
                }
            }
            // 6.更新消息统计文件
            Stat stat = readStat(msgQueue.getName());
            stat.totalCount = stat.totalCount + 1;
            stat.validCount = stat.validCount + 1;
            writeStat(msgQueue.getName(),stat);
        }
    }

    // 删除文件中的消息（逻辑删除）
    // 1.从文件中读取出消息（利用message中的offsetBeg、offsetEnd）
    // 2.将消息中的isVailid修改为0x0
    // 3.重新写入到原来的位置
    public void deleteMessage(MSGQueue msgQueue,Message message) throws IOException, ClassNotFoundException, MqException {
        synchronized (msgQueue) {
            // RandomAccessFile支持随机访问和读写文件.
            try(RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(msgQueue.getName()),"rw")) {
                // 1.从文件中读取message二进制数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                // 1.1 调整光标位置
                randomAccessFile.seek(message.getOffsetBeg());
                // 1.2 将消息读取到bufferSrc
                randomAccessFile.read(bufferSrc);
                // 2.将读取到的二进制数据反序列化为Message对象
                Message diskMessage = (Message) BinaryTool.toObject(bufferSrc);
                // 3.将isvalid修改为无效重新写入文件中(并不需要将内存中的isvalid设为0x0，因为内存中的即将删除)
                diskMessage.setIsVailid((byte) 0x0);
                byte[] invalidMessage = BinaryTool.toBytes(diskMessage);
                // 3.1 调整光标位置
                randomAccessFile.seek(message.getOffsetBeg());
                // 3.2 将逻辑删除的消息重新写回
                randomAccessFile.write(invalidMessage);
            }
            // 4.更新统计文件
            Stat stat = readStat(msgQueue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(msgQueue.getName(),stat);
        }
    }

    // 从文件中, 读取出所有有效消息内容.具体来说是放到一个链表里(方便后续头删操作)
    // 在程序启动时调用,不涉及多线程操作
    public LinkedList<Message> loadAllMessageFromFile(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        long gcBegTime = System.currentTimeMillis();

        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)) {

                // 记录当前文件光标位置
                long currentOffset = 0;
                while (true) {
                    // 1.获取当前消息的长度
                    int messageSize = dataInputStream.readInt();
                    // 2.将message二进制数据读取到buffer字节数组中
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    // 2.判断读取到的实际消息长度是否和文件中的消息长度一致
                    if (actualSize != messageSize) {
                        throw new MqException("文件格式错误: queueName = "+ queueName);
                    }
                    // 3.将读取到的消息反序列化为message对象
                    Message message = (Message) BinaryTool.toObject(buffer);
                    // 4.判定一下这个消息对象是否为有效对象
                    if (message.getIsVailid() != 0x1) {
                        currentOffset += (4 + actualSize);
                        // 无效数据直接跳过
                        continue;
                    }
                    // 5.有效数据,设置消息的offset,并将消息对象添加到链表中
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + actualSize);
                    messages.add(message);
                    currentOffset += (4 + actualSize);
                }
            } catch (EOFException e) {
                // 这个 catch 并非真是处理 "异常", 而是处理 "正常" 的业务逻辑. 文件读到末尾, 会被 readInt 抛出该异常.
                System.out.println("加载数据完成! queueDir = "+ getQueueDirPath(queueName));
            }
        }
        long gcEndTime = System.currentTimeMillis();

        System.out.println("加载数据共耗时: time = "+(gcEndTime-gcBegTime)+"ms.");

        return messages;
    }


    // ----------------------- GC操作相关 ---------------------------
    // 判断是否要对当前文件进行GC(条件：消息长度大于2000，有效消息比例小于50%)
    public boolean checkGC(String queueName) throws MqException {
        Stat stat = readStat(queueName);
        if (stat.totalCount > 2000 && (double)stat.validCount/stat.totalCount < 0.5) {
            return true;
        }
        return false;
    }

    // GC策略为复制删除，这里创建一个文件副本，便于将有效数据复制到副本中进行后续重命名
    private String getQueueDataNewPath(String queueName){
        return getQueueDirPath(queueName)+"/queue_data_new.txt";
    }

    // GC删除方法，用于垃圾回收
    public void gc(MSGQueue msgQueue) throws MqException, IOException, ClassNotFoundException {
        // 垃圾回收涉及修改操作，需要上锁
        synchronized (msgQueue) {
            long gcBegTime = System.currentTimeMillis();

            // 1.创建一个文件副本，用于存储有效消息
            File queueDataNewFile = new File(getQueueDataNewPath(msgQueue.getName()));
            // 1.1 判断文件副本是否已存在(如果已存在说明上次GC程序出错)
            if (queueDataNewFile.exists()) {
                throw new MqException("gc 时发现文件副本已存在: queueDataNewFile = "+queueDataNewFile.getAbsolutePath());
            }
            // 1.2 创建文件副本
            boolean res = queueDataNewFile.createNewFile();
            if (!res) {
                throw new IOException("创建文件副本失败: queueDataNewFile = "+queueDataNewFile.getAbsolutePath());
            }

            // 2.调用loadAllMessageFromFile()方法，获取所有信息
            LinkedList<Message> messages = loadAllMessageFromFile(msgQueue.getName());

            // 3.把有效消息写入到文件副本中
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)) {
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        // 先写入消息文件长度
                        dataOutputStream.writeInt(buffer.length);
                        // 再写入消息内容
                        dataOutputStream.write(buffer);
                    }
                }
            }

            // 4.删除旧的data文件，并将新data文件重命名
            File queueDataOldFile = new File(getQueueDataPath(msgQueue.getName()));
            // 4.1 删除文件
            res = queueDataOldFile.delete();
            if (!res) {
                throw new IOException("删除文件失败: queueDataOldFile = "+queueDataOldFile.getAbsolutePath());
            }
            // 4.2 文件重命名
            res = queueDataNewFile.renameTo(queueDataOldFile);
            if (!res) {
                throw new IOException("文件重命名失败: queueDataNewFile = " +queueDataNewFile.getAbsolutePath()+
                        " queueDataOldFile = "+queueDataOldFile.getAbsolutePath());
            }

            // 5.更新统计文件
            Stat stat = readStat(msgQueue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(msgQueue.getName(),stat);

            long gcEndTime = System.currentTimeMillis();
            System.out.println("gc 执行完毕. queueName = "+msgQueue.getName()+", time = "+(gcEndTime-gcBegTime)+" ms");
        }
    }


}
